package com.lzx.kafka.controller;

import com.alibaba.fastjson.JSON;
import com.lzx.kafka.helper.DingDingHelper;
import com.lzx.kafka.helper.EnterpriseWechat;
import com.lzx.kafka.helper.NotifyHelper;
import com.lzx.kafka.model.Notify;
import com.lzx.kafka.model.NotifyPlatform;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @ClassName KafkaReceiver
 * @Author 刘正星
 * @Date 2021/4/26 下午5:26
 * @Description
 */
@Component
public class KafkaReceiver {

    @Autowired
    private DingDingHelper dingDingHelper ;

    @Autowired
    private EnterpriseWechat enterpriseWechat;

    // #1. 监听主题为topic的消息
    @KafkaListener(topics = {"YouKeDa"})
    public void listen(ConsumerRecord<?, ?> record) {
        // #2. 如果消息存在
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            // #3. 获取消息
            String message = (String)kafkaMessage.get();
            Notify notify = JSON.parseObject(message,Notify.class);
            System.out.println(notify.getNotifyPlatform());
            if (notify.getNotifyPlatform().equals(NotifyPlatform.DINGDING)){
              boolean dingResult =  dingDingHelper.sendNotify(notify);
                System.out.println(dingResult);
            }else{
                enterpriseWechat.sendNotify(notify);
            }
            System.out.println("message =" + message);
        }

    }
}
